package com.dtwave.cfstreaming.connection;

import org.apache.commons.lang.StringUtils;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

public class ElasticConnector {
    private final static Logger LOGGER = LoggerFactory.getLogger(ElasticConnector.class);

    //volatile可以被修改的变脸，其他占用的线程的数据可以被修改
    private static volatile RestHighLevelClient client = null;

    /**
     * 获取全局唯一的client
     */
    public static synchronized RestHighLevelClient getClient(String url) {
        if (client == null) {
            LOGGER.info("获取全局唯一client实例");
            List<HttpHost> hosts = parse(url);
            if (hosts.size() == 3) {
                client = new RestHighLevelClient(RestClient.builder(hosts.get(0), hosts.get(1), hosts.get(2)));
            } else if (hosts.size() == 2) {
                client = new RestHighLevelClient(RestClient.builder(hosts.get(0), hosts.get(1)));
            } else if (hosts.size() == 1) {
                client = new RestHighLevelClient(RestClient.builder(hosts.get(0)));
            } else {
                LOGGER.info("esUrl解析失败！");
            }
        }
        return client;
    }

    /**
     * 解析es地址
     */
    public static List<HttpHost> parse(String connUrl) {
        String[] esNodes = connUrl.split(",");
        List<HttpHost> esSocketList = new ArrayList<>();
        for (String esNode : esNodes) {
            esNode = esNode.startsWith("http://") ? StringUtils.substringAfter(esNode, "http://") : esNode;
            String[] hostAndPort = esNode.split(":");
            if (!StringUtils.isNumeric(hostAndPort[1])) {
                throw new IllegalArgumentException("the port must be numberic");
            }
            //抽取host
            esSocketList.add(new HttpHost(hostAndPort[0], Integer.valueOf(hostAndPort[1]), "http"));
        }
        return esSocketList;
    }

}
